package com.zto.wk.webflux.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * Created by wangkuan at 2018-08-14 16:17
 */
public class KafkaUtil {
    private static final String BROKER_LIST = "test.server:9092";
    private static KafkaProducer<String, String> producer;

    public KafkaUtil(Properties prop) {
        if (prop == null) {
            prop = new Properties();
            prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_LIST);
            prop.put(ProducerConfig.ACKS_CONFIG, "-1");
        }

        prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        producer = new KafkaProducer<>(prop);
    }

    public void send(String topic, String value) {
        producer.send(new ProducerRecord<>(topic, value));

        producer.flush();
    }

    public static void main(String[] args) {
        MyUtil.myUtil.INSTANCE.print("好好好");
    }
}
